package com.dabai.iot.core.dubbo;

import com.alibaba.fastjson2.JSONObject;
import com.alibaba.fastjson2.JSONValidator;
import com.dabai.common.core.enums.YnEnum;
import com.dabai.common.core.utils.Func;
import com.dabai.common.core.utils.SpringUtils;
import com.dabai.iot.common.enums.*;
import com.dabai.iot.common.model.DeviceInfo;
import com.dabai.iot.common.model.IotMqttMessage;
import com.dabai.iot.common.model.IotMqttStatus;
import com.dabai.iot.core.api.RemoteMessageHandlerService;
import com.dabai.iot.core.entity.ConsumerGroupEntity;
import com.dabai.iot.core.entity.DeviceEntity;
import com.dabai.iot.core.entity.ProductEntity;
import com.dabai.iot.core.entity.ServerSubscribeEntity;
import com.dabai.iot.core.event.AlarmEvent;
import com.dabai.iot.core.service.*;
import com.dabai.iot.core.task.DeviceStatusProcessor;
import com.dabai.iot.core.utils.ServerMessageHandler;
import com.dabai.iot.core.utils.TslMessageParser;
import com.dabai.iot.mqtt.api.RemoteMqttService;
import com.dabai.iot.mqtt.topics.SysTopics;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.dubbo.config.annotation.DubboReference;
import org.apache.dubbo.config.annotation.DubboService;
import org.springframework.stereotype.Service;
import org.springframework.util.StopWatch;

import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;

/**
 * 消息处理服务
 *
 * @author kev1n
 */
@Slf4j
@Service
@DubboService
@RequiredArgsConstructor
public class RemoteMessageHandlerServiceImpl implements RemoteMessageHandlerService {

    private final IProductService productService;
    private final IDeviceService deviceService;
    private final IDeviceLogService deviceLogService;
    private final IDeviceCountService deviceCountService;
    private final IDevicePropertyService devicePropertyService;
    private final IServerSubscribeService serverSubscribeService;
    private final IConsumerGroupService consumerGroupService;
    private final DeviceStatusProcessor deviceStatusProcessor;

    @DubboReference
    private final RemoteMqttService remoteMqttService;

    /**
     * 设备行为（上线、离线）处理
     */
    @Override
    public void deviceStatusHandler(DeviceInfo deviceInfo, IotMqttStatus mqttStatus) {
        StopWatch stopWatch = new StopWatch("deviceStatusHandler");
        stopWatch.start("getDeviceCacheBySn");

        DeviceEntity device = deviceService.getDeviceCacheBySn(deviceInfo.productKey(), deviceInfo.deviceSn());
        if (device == null) {
            return;
        }

        stopWatch.stop();
        stopWatch.start("online");

        if (ConnectStatus.online.equals(mqttStatus.status())) {
            device.setIsOnline(YnEnum.YES.getValue());
            // 更新缓存中当天在线设备集合
            deviceCountService.updateDeviceOnlineCount(deviceInfo.tenantId(), mqttStatus.reportTime(), Set.of(deviceInfo.deviceId()));
        } else if (ConnectStatus.offline.equals(mqttStatus.status())) {
            device.setIsOnline(YnEnum.NO.getValue());
        }

        stopWatch.stop();
        stopWatch.start("saveDeviceStatusLogToTD");
        // 保存状态日志
        deviceLogService.saveDeviceStatusLogToTD(deviceInfo, mqttStatus);

        stopWatch.stop();
        stopWatch.start("deviceStatusProcessor");
        // 更新设备状态
        deviceStatusProcessor.add(device);

        stopWatch.stop();
        stopWatch.start("getConsumerGroupCacheByProductId");

        // 组装服务端订阅使用的设备状态数据
        JSONObject returnJsonObject = ServerMessageHandler.assembleStatusToServerPayload(
            deviceInfo.productKey(),
            deviceInfo.deviceSn(),
            mqttStatus.status().name(),
            mqttStatus.reportTime()
        );

        // 判断是否有服务端订阅
        if (hasServerSubscribe(deviceInfo.productId(), MessageType.status)) {
            sendMessageToServerByMQTT(mqttStatus.traceId(), deviceInfo.tenantId(), deviceInfo.productId(), MessageType.status, returnJsonObject, stopWatch);
        }

        stopWatch.stop();
        stopWatch.start("AlarmEvent");

        // 告警监听
        AlarmEvent alarmEvent = new AlarmEvent();
        alarmEvent.setTraceId(mqttStatus.traceId());
        alarmEvent.setDeviceId(device.getId());
        alarmEvent.setConditionEnum(AlarmConditionEnum.STATUS);
        alarmEvent.setStatusValue(mqttStatus.status().name());
        SpringUtils.publishEvent(alarmEvent);

        stopWatch.stop();
        log.info(stopWatch.prettyPrint(TimeUnit.MILLISECONDS));
        log.info("StopWatch ---===> [{}]: running time = {} ms", stopWatch.getId(), stopWatch.getTotalTimeMillis());
    }

    /**
     * 设备消息处理
     */
    @Override
    public void deviceMessageHandler(DeviceInfo deviceInfo, IotMqttMessage mqttMessage) {
        StopWatch stopWatch = new StopWatch("deviceMessageHandler");
        stopWatch.start("getProductCacheByPk");

        ProductEntity product = productService.getProductCacheByPk(deviceInfo.productKey());
        if (product == null) {
            return;
        }

        stopWatch.stop();
        stopWatch.start("onlySaveLog");

        String tsl = product.getTsl();
        String upPayload = mqttMessage.payload();
        String finalPayload = upPayload;
        MessageType messageType = MessageType.property;

        // 验证 payload 格式，如果格式不正确，则只记录日志
        BizLogCode logCode = validatePayloadFormat(upPayload);
        if (!BizLogCode.SUCCESS.equals(logCode)) {
            deviceLogService.saveDeviceLogToTD(deviceInfo, mqttMessage, BizLogType.uplink, logCode);
            return;
        }

        stopWatch.stop();
        stopWatch.start("savePropertiesLogToTD");

        if (isSysPropertyPostTopic(deviceInfo.productKey(), deviceInfo.deviceSn(), mqttMessage.topic())) {

            if (Func.isNotBlank(tsl)) {
                // 将符合物模型的消息按照物模型TSL进行解析
                JSONObject tslMessage = TslMessageParser.parseThingModelMessage(mqttMessage.reportTime(), upPayload, tsl);
                finalPayload = tslMessage.toString();

                // 将设备上报的属性消息存储到 redis
                devicePropertyService.saveLastPropertiesToRedis(deviceInfo.deviceId(), finalPayload);

                // 记录属性日志
                deviceLogService.savePropertiesLogToTD(deviceInfo.productKey(), deviceInfo.deviceSn(), tslMessage.toString());
            }

        } else {
            messageType = MessageType.message;
        }

        stopWatch.stop();
        stopWatch.start("saveDeviceLogToTD");

        // 设备日志保存原始上报报文
        deviceLogService.saveDeviceLogToTD(deviceInfo, mqttMessage, BizLogType.uplink, logCode);

        stopWatch.stop();
        stopWatch.start("getConsumerGroupCacheByProductId");

        // 组装服务端订阅使用的设备消息数据
        JSONObject returnJsonObject = ServerMessageHandler.assembleMessageToServerPayload(deviceInfo, mqttMessage, finalPayload);

        // 判断是否有服务端订阅
        if (hasServerSubscribe(deviceInfo.productId(), messageType)) {
            sendMessageToServerByMQTT(mqttMessage.traceId(), deviceInfo.tenantId(), deviceInfo.productId(), messageType, returnJsonObject, stopWatch);
        }

        stopWatch.stop();
        stopWatch.start("AlarmEvent");

        // 告警监听
        AlarmEvent alarmEvent = new AlarmEvent();
        alarmEvent.setDeviceId(deviceInfo.deviceId());
        alarmEvent.setTraceId(mqttMessage.traceId());
        alarmEvent.setConditionEnum(AlarmConditionEnum.PROPERTY);
        alarmEvent.setPropertyPayload(finalPayload);
        SpringUtils.publishEvent(alarmEvent);

        stopWatch.stop();
        log.info(stopWatch.prettyPrint(TimeUnit.MILLISECONDS));
        log.info("StopWatch ---===> [{}]: running time = {} ms", stopWatch.getId(), stopWatch.getTotalTimeMillis());
    }

    /**
     * 是否物模型属性上报
     */
    private boolean isSysPropertyPostTopic(String productKey, String deviceSn, String reportTopic) {
        // 系统Topic - 设备上报属性数据
        String sysTopicPropertyReport = SysTopics.PROPERTY_POST
            .replace("${productKey}", productKey)
            .replace("${deviceSn}", deviceSn);

        return sysTopicPropertyReport.equals(reportTopic);
    }

    /**
     * 验证 payload 格式
     */
    private BizLogCode validatePayloadFormat(String payload) {
        JSONValidator validator = JSONValidator.from(payload);
        if (!validator.validate() ||
            validator.getType() == JSONValidator.Type.Array ||
            validator.getType() == JSONValidator.Type.Value) {
            return BizLogCode.INVALID_MESSAGE_FORMAT;
        }
        return BizLogCode.SUCCESS;
    }

    /**
     * 将组装好的需要发送到服务端的消息通过 MQTT 模块发送出去
     */
    private void sendMessageToServerByMQTT(String traceId, String tenantId, Long productId, MessageType messageType, JSONObject jsonObject, StopWatch stopWatch) {
        // 查询产品对应消费组（一对多）
        List<ConsumerGroupEntity> consumerGroupList = consumerGroupService.getCacheByProductId(productId);

        stopWatch.stop();
        stopWatch.start("sendMessageToServerByMQTT");

        if (Func.isNotEmpty(consumerGroupList)) {
            log.info("服务端订阅 ======> 当前产品有【{}】个消费组", consumerGroupList.size());
            for (ConsumerGroupEntity consumerGroup : consumerGroupList) {
                String groupID = consumerGroup.getGroupIdentifier();
                String topic = MessageType.status.equals(messageType)
                    ? SysTopics.getStatusTopic(tenantId, groupID)
                    : SysTopics.getDataTopic(tenantId, groupID);

                remoteMqttService.sendMessageToServer(traceId, consumerGroup.getId(), groupID, topic, jsonObject.toString());
            }
        }
    }

    /**
     * 判断是否有服务端订阅
     */
    private boolean hasServerSubscribe(Long productId, MessageType reportType) {
        // 查询设备产品对应的订阅信息
        ServerSubscribeEntity serverSubscribe = serverSubscribeService.getCacheByProductId(productId);
        if (serverSubscribe == null) {
            return false;
        }

        String messageType = serverSubscribe.getMessageType();
        if (Func.isBlank(messageType)) {
            return false;
        }

        // 订阅类型和上报类型是否匹配
        return messageType.contains(String.valueOf(reportType.getValue()));
    }

    /**
     * 服务端发布消息处理（调用 iot 模块只为记录日志）
     */
    @Override
    public void serverMessageHandler(DeviceInfo deviceInfo, IotMqttMessage mqttMessage) {
        deviceLogService.saveDeviceLogToTD(deviceInfo, mqttMessage, BizLogType.downLink, BizLogCode.SUCCESS);
    }

    /**
     * 服务端订阅应答消息处理（调用 iot 模块只为记录日志）
     */
    @Override
    public void serverAckMessageHandler(DeviceInfo deviceInfo, IotMqttMessage mqttMessage) {
        deviceLogService.saveDeviceLogToTD(deviceInfo, mqttMessage, BizLogType.serverSubscribe, BizLogCode.SUCCESS);
    }
}
